package org.poem.exec.handler;

import org.poem.CanalApplicationContext;
import org.poem.CanalClientHeadEntityVO;
import org.poem.ExecTaskDetailPlanVO;
import org.poem.TaskDetailAvailabilityMaps;
import org.poem.canal.CanalHandleEntry;
import org.poem.handler.full.AbstractClientFullHandler;
import org.poem.handler.increment.AbstractClientIncrementHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * * 定时任务的执行
 * * 把任务添加进来就可以执行
 * * 这个是spring quartz自动执行的
 *
 * @author poem
 */
public class ClientHandlerScheduledExecutor {

    /**
     * 日志管理
     */
    private static final Logger logger = LoggerFactory.getLogger(ClientHandlerScheduledExecutor.class);

    private static ExecutorService threadPoolExecutor;


    static {
        int core = Runtime.getRuntime().availableProcessors();
        AtomicInteger integer = new AtomicInteger(0);
        threadPoolExecutor = new ThreadPoolExecutor(core, core * 2, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(core),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        //线程命名
                        return new Thread(r, "full-schedule-" + integer.incrementAndGet());
                    }
                }, new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /**
     * 制定操作
     *
     * @param config
     */
    private static void execute(CanalHandleEntry config, CanalClientHeadEntityVO entity) {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    Class clazz;
                    Class parentClass;
                    String taskClass = config.getClassName();
                    if (!StringUtils.isEmpty(taskClass)) {
                        Object clsObj = CanalApplicationContext.getBean(config.getAClass());
                        clazz = clsObj.getClass();
                        while (true) {
                            parentClass = clazz.getSuperclass();
                            if (parentClass == null) {
                                break;
                            }
                            if (clazz.getSuperclass().getName().equals(AbstractClientIncrementHandler.class.getName())) {
                                AbstractClientIncrementHandler abstractMonitorService = (AbstractClientIncrementHandler) clsObj;
                                abstractMonitorService.handle(entity);
                                break;
                            }
                            clazz = parentClass;
                        }
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        });
    }

    /**
     * 制定操作
     *
     * @param config
     */
    private static void execute(CanalHandleEntry config, ExecTaskDetailPlanVO entity) {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    Integer availability = TaskDetailAvailabilityMaps.getAvailability(entity.getTargetTable());
                    if (availability == 1) {
                        Class clazz;
                        Class parentClass;
                        String taskClass = config.getClassName();
                        if (!StringUtils.isEmpty(taskClass)) {
                            Object clsObj = CanalApplicationContext.getBean(config.getAClass());
                            clazz = clsObj.getClass();
                            while (true) {
                                availability = TaskDetailAvailabilityMaps.getAvailability(entity.getTargetTable());
                                if (availability == 1) {
                                    parentClass = clazz.getSuperclass();
                                    if (parentClass == null) {
                                        break;
                                    }
                                    if (clazz.getSuperclass().getName().equals(AbstractClientFullHandler.class.getName())) {
                                        AbstractClientFullHandler abstractMonitorService = (AbstractClientFullHandler) clsObj;
                                        abstractMonitorService.handle(entity);
                                        break;
                                    }
                                    clazz = parentClass;
                                } else {
                                    return;
                                }

                            }
                        }
                    }

                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        });
    }
    /**
     * 队列执行
     *
     * @param canalHandleEntries
     * @param entity
     */
    public static void handlerIncrementConsumer(List<CanalHandleEntry> canalHandleEntries, CanalClientHeadEntityVO entity) {
        for (CanalHandleEntry canalHandleEntry : canalHandleEntries) {
            execute(canalHandleEntry, entity);
        }
    }
    /**
     * 队列执行
     *
     * @param canalHandleEntries
     * @param entity
     */
    public static void handlerFullConsumer(List<CanalHandleEntry> canalHandleEntries, ExecTaskDetailPlanVO entity) {
        for (CanalHandleEntry canalHandleEntry : canalHandleEntries) {
            execute(canalHandleEntry, entity);
        }
    }
}
